package com.practice.gmall.realtime.app.dwd;

import com.practice.gmall.realtime.app.BaseSQLApp;
import com.practice.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DWD_04_TradeOrderCancelDetail
 * Package: com.practice.gmall.realtime.app.dwd
 * Description:
 * <p>
 * 交易域取消订单明细表
 * <p>
 * 粒度  商品粒度
 * 维度  商品、省份、
 * 度量  订单金额、优惠券金额、活动金额
 * <p>
 * 关联的表  oder_info order_detail order_detail_activity order_detail_coupon base_dic
 *
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-14 14:27
 */
public class DWD_04_TradeOrderCancelDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_04_TradeOrderCancelDetail().init(3338, "DWD_04_TradeOrderCancelDetail", 2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. read ods_db
        //MyKafkaUtil.readTopicOdsDb(tEnv, "DWD_04_TradeOrderCancelDetail");

        // 2. read base_dic lookup table
        //tEnv.executeSql(MySQLUtil.getBaseDicLookUp());

        // 3. read relative table;
        // 3.1 main table order_detail(from Kafka topic dwd_order_detail)
        tEnv.executeSql("" +
                "CREATE TABLE dwd_order_detail( " +
                "    id  string,    " +
                "    order_id  string,    " +
                "    user_id  string,   " +
                "    sku_id  string,    " +
                "    sku_name  string,    " +
                "    order_price  string,    " +
                "    sku_num  string,    " +
                "    date_id  string,    " +
                "    province_id  string, " +
                "    source_type_code  string, " +
                "    source_type_name  string, " +
                "    source_id  string, " +
                "    activity_id  string, " +
                "    activity_rule_id   string, " +
                "    coupon_id  string, " +
                "    split_original_amount  string,    " +
                "    split_total_amount   string, " +
                "    split_activity_amount   string, " +
                "    split_coupon_amount   string, " +
                "    `ts` BIGINT ," +
                "     pt as PROCTIME()" +
                ")" +
                MyKafkaUtil.getKafkaSourceDDL("dwd_order_detail","DWD_04_TradeOrderCancelDetail"));


        // 3.2 order_info(status : 1003 )
        // analyse: 1. canceled <- status:1003 
        Table canceled_order_info = tEnv.sqlQuery("" +
                "SELECT " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['consignee'] consignee, " +
                "    `data`['province_id'] province_id, " +
                "    `data`['order_status'] order_status, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['operate_time'] cancel_time, " +
                "    `data`['process_status'] process_status, " +
                "    `data`['parent_order_id'] parent_order_id " +
                "FROM ods_db " +
                "WHERE " +
                "    `database`='gmall2022'   " +
                "    AND `table`='order_info'   " +
                "    AND `type`='update' " +
                "    AND `old`['order_status'] = '1001' " +
                "    AND `data`['order_status'] = '1003'");

        tEnv.createTemporaryView("order_info", canceled_order_info);


        // 4. join table
        Table result = tEnv.sqlQuery("" +
                "SELECT " +
                "   dod.id , " +
                "   dod.order_id , " +
                "   oi.user_id , " +
                "   dod.sku_id , " +
                "   dod.sku_name , " +
                "   order_price , " +
                "   dod.sku_num , " +
                "   dod.date_id , " +
                "   oi.province_id , " +
                "   dod.source_type_code , " +
                "   dic.dic_name source_type_name , " +
                "   dod.source_id , " +
                "   dod.activity_id , " +
                "   dod.activity_rule_id  , " +
                "   dod.coupon_id , " +
                "   dod.split_original_amount , " +
                "   dod.split_total_amount  , " +
                "   dod.split_activity_amount  , " +
                "   dod.split_coupon_amount  , " +
                "   dod.`ts`     " +
                " " +
                "FROM  " +
                " " +
                "   dwd_order_detail dod JOIN  order_info oi on oi.id = dod.order_id ");
                /* 相关维表已经包含在dwd_order_detail, 其他两个表不再需要关联
                +
                "   JOIN base_dic FOR SYSTEM_TIME AS OF dod.pt AS dic " +
                "   on dic.dic_code = dod.source_type_code");
                */


        // 5. create table dwd_order_cancel
        tEnv.executeSql("" +
                "CREATE TABLE dwd_order_canceled(   " +
                "    id  string,      " +
                "    order_id  string,      " +
                "    user_id  string,     " +
                "    sku_id  string,      " +
                "    sku_name  string,      " +
                "    order_price  string,      " +
                "    sku_num  string,      " +
                "    date_id  string,      " +
                "    province_id  string,   " +
                "    source_type_code  string,   " +
                "    source_type_name  string,   " +
                "    source_id  string,   " +
                "    activity_id  string,   " +
                "    activity_rule_id   string,   " +
                "    coupon_id  string,   " +
                "    split_original_amount  string,      " +
                "    split_total_amount   string,   " +
                "    split_activity_amount   string,   " +
                "    split_coupon_amount   string,   " +
                "    `ts` BIGINT   " +
                ")  " +
                MyKafkaUtil.getKafkaSinkDDL("dwd_order_canceled"));


        // 6. write to Kafka
        tEnv.executeSql("insert into dwd_order_canceled select * from " + result);
    }
}
